package com.thiswhyme.republish.rabbitmq.datasource;

import com.thiswhyme.common.expection.BusinessException;
import com.thiswhyme.republish.common.utils.ResultSetUtils;
import com.thiswhyme.republish.rabbitmq.core.domain.MessageInfo;
import com.thiswhyme.republish.rabbitmq.core.domain.PublishStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.*;
import java.time.Instant;
import java.util.List;

/**
 * MessageInfo仓储服务
 * Created by thiswhyme on 2021-03-21.
 */
public class MessageInfoRepo {

    public static final Logger log = LoggerFactory.getLogger(MessageInfoRepo.class);

    public MessageInfoRepo(DataSource ds, MessageInfoSqlBuilder misb) {
        this.dataSource = ds;
        this.messageInfoSqlBuilder = misb;
    }

    private DataSource dataSource;

    private MessageInfoSqlBuilder messageInfoSqlBuilder;

    /**
     * 新增MessageInfo
     * @param msg
     * @throws SQLException
     */
    public int insert(MessageInfo msg) {
        Connection conn = null;
        PreparedStatement pst = null;
        try {
            conn = dataSource.getConnection();
            String sql = messageInfoSqlBuilder.buildInsertSql("");
            pst = conn.prepareStatement(sql);
            pst.setLong(1, msg.getId());
            pst.setInt(2, msg.getType());
            pst.setString(3, msg.getChannelName());
            pst.setString(4, msg.getMsgId());
            pst.setString(5, msg.getHeaders());
            pst.setString(6, msg.getContent());
            pst.setString(7, msg.getContentClassType());
            pst.setInt(8, msg.getPublishStatus());
            pst.setInt(9, msg.getRetryTimes());
            pst.setTimestamp(10, msg.getCreateTime() == null ? Timestamp.from(Instant.now()) : msg.getCreateTime());
            pst.setTimestamp(11, null);
            return pst.executeUpdate();
        } catch (Exception ex) {
            log.error("执行insert失败，参数：{}", msg.toString(), ex);
            throw new BusinessException("消息记录失败");
        } finally {
            closeConnection(conn, pst);
        }
    }

    /**
     * 更新重试次数
     * @param dataId
     * @param retryTimes
     * @return
     */
    public int updateRetryTimes(Long dataId, int retryTimes) {
        Connection conn = null;
        PreparedStatement pst = null;
        try {
            conn = dataSource.getConnection();
            String sql = messageInfoSqlBuilder.buildUpdateRetryTimesSql();
            pst = conn.prepareStatement(sql);
            pst.setInt(1, retryTimes);
            pst.setTimestamp(2, Timestamp.from(Instant.now()));
            pst.setLong(3, dataId);
            return pst.executeUpdate();
        } catch (Exception ex) {
            log.error("执行updateRetryTimes失败，参数：{}", dataId, ex);
            throw new BusinessException("消息重发次数更新失败");
        } finally {
            closeConnection(conn, pst);
        }
    }

    /**
     * 更新发送状态
     * @param dataId
     * @param status
     * @return
     */
    public int updatePublishStatus(Long dataId, PublishStatus status) {
        Connection conn = null;
        PreparedStatement pst = null;
        try {
            conn = dataSource.getConnection();
            String sql = messageInfoSqlBuilder.buildUpdateStatusSql();
            pst = conn.prepareStatement(sql);
            pst.setInt(1, status.getValue());
            pst.setTimestamp(2, Timestamp.from(Instant.now()));
            pst.setLong(3, dataId);
            return pst.executeUpdate();
        } catch (Exception ex) {
            log.error("执行updatePublishStatus失败，参数：{}", dataId, ex);
            throw new BusinessException("消息发送状态更新失败");
        } finally {
            closeConnection(conn, pst);
        }
    }

    /**
     * 批量删除
     * @param dataIds
     * @return
     */
    public int batchDelete(List<Long> dataIds) {
        Connection conn = null;
        PreparedStatement pst = null;
        try {
            conn = dataSource.getConnection();
            String sql = messageInfoSqlBuilder.buildDeleteSql(dataIds, "");
            pst = conn.prepareStatement(sql);
            for (int i = 1; i <= dataIds.size(); i++) {
                pst.setLong(i, dataIds.get(i - 1));
            }
            return pst.executeUpdate();
        } catch (Exception ex) {
            log.error("执行delete失败，参数：{}", dataIds, ex);
            throw new BusinessException("批量删除消息失败");
        } finally {
            closeConnection(conn, pst);
        }
    }

    /**
     * 根据ID列表查询消息
     * @param dataIds
     * @return
     */
    public List<MessageInfo> selectByIds(List<Long> dataIds) {
        Connection conn = null;
        PreparedStatement pst = null;
        ResultSet rs = null;
        try {
            conn = dataSource.getConnection();
            String sql = messageInfoSqlBuilder.buildSelectSql(dataIds);
            pst = conn.prepareStatement(sql);
            for (int i = 1; i <= dataIds.size(); i++) {
                pst.setLong(i, dataIds.get(i - 1));
            }
            rs = pst.executeQuery();
            return ResultSetUtils.resultSetToList(rs, MessageInfo.class);
        } catch (Exception ex) {
            log.error("执行select失败，参数：{}", dataIds, ex);
            throw new BusinessException("消息查询失败");
        } finally {
            closeConnection(conn, pst, rs);
        }
    }

    /**
     * 查询指定时间之前创建的指定状态的消息
     * @param statusList
     * @param limitTime
     * @param start
     * @param limit
     * @return
     */
    public List<MessageInfo> selectByCreateTime(List<Integer> statusList, Timestamp limitTime, int start, int limit) {
        Connection conn = null;
        PreparedStatement pst = null;
        ResultSet rs = null;
        try {
            conn = dataSource.getConnection();
            String sql = messageInfoSqlBuilder.buildSelectByCreateTimeSql(statusList);
            pst = conn.prepareStatement(sql);
            for (int i = 1; i <= statusList.size(); i++) {
                pst.setInt(i, statusList.get(i - 1));
            }
            pst.setTimestamp(statusList.size() + 1, limitTime);
            pst.setInt(statusList.size() + 2, start);
            pst.setInt(statusList.size() + 3, limit);
            rs = pst.executeQuery();
            return ResultSetUtils.resultSetToList(rs, MessageInfo.class);
        } catch (Exception ex) {
            log.error("执行selectByCreateTime失败，参数：{}，{}", statusList, limitTime, ex);
            throw new BusinessException("消息查询失败");
        } finally {
            closeConnection(conn, pst, rs);
        }
    }

    /**
     * 关闭连接
     * @param conn
     * @param pst
     */
    private void closeConnection(Connection conn, PreparedStatement pst) {
        try {
            if (pst != null) {
                pst.close();
            }
            if (conn != null) {
                conn.close();
            }
        } catch (SQLException closeEx) {
            log.warn("关闭JDBC连接失败 ... ", closeEx);
        }
    }

    /**
     * 关闭连接
     * @param conn
     * @param pst
     * @param rs
     */
    private void closeConnection(Connection conn, PreparedStatement pst, ResultSet rs) {
        try {
            if (rs != null) {
                rs.close();
            }
            if (pst != null) {
                pst.close();
            }
            if (conn != null) {
                conn.close();
            }
        } catch (SQLException closeEx) {
            log.warn("关闭JDBC连接失败 ... ", closeEx);
        }
    }
}
